package com.atguigu.flink.tableapi;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.*;

/**
 * Created by Smexy on 2023/2/5
 */
public class Demo10_GroupWindow
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        env.setParallelism(1);

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner((e, ts) -> e.getTs());


        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy);


        /*
                只选取流中数据的某些属性，组成Table
         */
        Table table = tableEnvironment.fromDataStream(ds, $("id"),$("ts"),$("vc"),
            $("pt").proctime(),$("et").rowtime()
        );

        /*
                两种聚合:
                    groupWindow:  flink中提供的窗口功能。
                    overWindow:   SQL中提供的开窗函数。
                                    select  窗口函数() over( partition by xx order by xx window范围 )
                 -------------
   Table table = input
  .window([GroupWindow w].as("w"))  // define window with alias w
  .groupBy($("w"))  // group the table by window w
  .select($("b").sum());  // aggregate
         */

        //构造滚动窗口
        /*
                Tumbling Event-time Window
                    over: 窗口长度，需要传入一个字面量
                    on: 窗口类型，基于时间还是元素个数
                    as: 起别名

                Sliding Window:

                    every： 滑动步长
         */
        TumbleWithSizeOnTimeWithAlias w1 = Tumble.over(lit(5).seconds()).on($("et")).as("w");
        //Processing-time Window
        TumbleWithSizeOnTimeWithAlias w2 = Tumble.over(lit(5).seconds()).on($("pt")).as("w");
        //元素个数  必须制定处理时间，需要按照处理时间将数据进行排序
        TumbleWithSizeOnTimeWithAlias w3 = Tumble.over(rowInterval(3l)).on($("pt")).as("w");

        //构造滑动窗口
        SlideWithSizeAndSlideOnTimeWithAlias w4 = Slide.over(lit(5).seconds()).every(lit(3).seconds()).on($("et")).as("w");
        SlideWithSizeAndSlideOnTimeWithAlias w5 = Slide.over(lit(5).seconds()).every(lit(3).seconds()).on($("pt")).as("w");
        //sql： 第一次触发必须满足 窗口大小，才会计算，之后都是没slide算一次
        SlideWithSizeAndSlideOnTimeWithAlias w6 = Slide.over(rowInterval(5l)).every(rowInterval(3l)).on($("pt")).as("w");

             //会话窗口
        SessionWithGapOnTimeWithAlias w7 = Session.withGap(lit(3).seconds()).on($("et")).as("w");
        SessionWithGapOnTimeWithAlias w8 = Session.withGap(lit(3).seconds()).on($("pt")).as("w");

        /*
                不keyBy，也称为全局窗口。 所有的key都是进入同一个窗口。
                keyBy，需要在groupBy($("w"),key)
         */
        table.window(w3)
             .groupBy($("w"),$("id"))
             //不能使用w作为查询的字段，如果是时间窗口，可以使用 $("w").start()获取窗口的起始时间，和$("w").end()获取窗口的结束时间
             //.select($("w").start().as("wstart"),$("w").end().as("wend"),$("vc").sum().as("sumVC"))
             //基于个数的窗口，无法获取窗口的属性
             .select($("id"),$("vc").sum().as("sumVC"))
             .execute()
             .print();


    }
}
